package com.atguigu.dga.governance.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.ds.bean.TDsTaskDefinition;
import com.atguigu.dga.ds.bean.TDsTaskInstance;
import com.atguigu.dga.ds.service.TDsTaskDefinitionService;
import com.atguigu.dga.ds.service.TDsTaskInstanceService;
import com.atguigu.dga.governance.assessor.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.governance.bean.GovernanceMetric;
import com.atguigu.dga.governance.mapper.GovernanceAssessDetailMapper;
import com.atguigu.dga.governance.service.GovernanceAssessDetailService;
import com.atguigu.dga.governance.service.GovernanceMetricService;
import com.atguigu.dga.meta.bean.TableMetaInfo;
import com.atguigu.dga.meta.mapper.TableMetaInfoMapper;
import com.atguigu.dga.meta.service.TableMetaInfoExtraService;
import com.atguigu.dga.meta.service.TableMetaInfoService;
import com.atguigu.dga.util.SpringIOCProvider;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * <p>
 * 治理考评结果明细 服务实现类
 * </p>
 *
 * @author wuzhida
 * @since 2024-06-26
 */
@Service
@DS("dga")
public class GovernanceAssessDetailServiceImpl extends ServiceImpl<GovernanceAssessDetailMapper, GovernanceAssessDetail> implements GovernanceAssessDetailService {

    @Autowired
    private TableMetaInfoService tableMetaInfoService;
    @Autowired
    private TableMetaInfoExtraService tableMetaInfoExtraService;
    @Autowired
    private TableMetaInfoMapper tableMetaInfoMapper;
    @Autowired
    private GovernanceMetricService governanceMetricService;
    @Autowired
    private SpringIOCProvider springIOCProvider;
    @Autowired
    private TDsTaskDefinitionService tDsTaskDefinitionService;
    @Autowired
    private TDsTaskInstanceService tDsTaskInstanceService;
    
    //线程池
    private ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(6, 6, 10, TimeUnit.MINUTES, new LinkedBlockingDeque<>(1600));
    
    /**
     * 考评流程: 
     * 0. 删除当前考评日期的考评信息
     * 1. 获取所有的表信息
     * 2. 获取所有的指标信息
     * 3. 考评
     * 4. 返回考评结果
     * @param assessDate 考评日期
     */
    @Override
    public void mainAssess(String assessDate){
        //0. 删除当前考评日期的考评信息
        remove(
                new QueryWrapper<GovernanceAssessDetail>()
                        .eq("assess_date", assessDate)
        );
        
        //1. 获取所有的表信息
        List<TableMetaInfo> tableMetaInfoList = tableMetaInfoMapper.selectAllTableMetaInfoWithExtra(assessDate);
//        List<TableMetaInfo> tableMetaInfoList = tableMetaInfoService.list(
//                new QueryWrapper<TableMetaInfo>()
//                        .eq("assess_date", assessDate)
//        );
//        for (TableMetaInfo tableMetaInfo : tableMetaInfoList) {
//            TableMetaInfoExtra tableMetaInfoExtra = tableMetaInfoExtraService.getOne(
//                    new QueryWrapper<TableMetaInfoExtra>()
//                            .eq("schema_name", tableMetaInfo.getSchemaName())
//                            .eq("table_name", tableMetaInfo.getTableName())
//            );
//            tableMetaInfo.setTableMetaInfoExtra(tableMetaInfoExtra);
//        }

        //2. 获取所有的指标信息
        List<GovernanceMetric> governanceMetricList = governanceMetricService.list(
                new QueryWrapper<GovernanceMetric>()
                        .eq("is_disabled", "0")
        );
        System.out.println(governanceMetricList);
        
        //从dolphinscheduler获取任务定义和任务实例
        //任务定义
        List<TDsTaskDefinition> tDsTaskDefinitionList = tDsTaskDefinitionService.getTDsTaskDefinitionList(assessDate);
        Map<String, TDsTaskDefinition> tDsTaskDefinitionMap = new HashMap<>();
        for (TDsTaskDefinition tDsTaskDefinition : tDsTaskDefinitionList) {
            tDsTaskDefinitionMap.put(tDsTaskDefinition.getName(), tDsTaskDefinition);
        }
        //任务实例
        List<TDsTaskInstance> tDsTaskInstanceList = tDsTaskInstanceService.getTDsTaskInstanceList(assessDate);
        Map<String, TDsTaskInstance> tDsTaskInstanceMap = new HashMap<>();
        for (TDsTaskInstance tDsTaskInstance : tDsTaskInstanceList) {
            tDsTaskInstanceMap.put(tDsTaskInstance.getName(), tDsTaskInstance);
        }

        long start = System.currentTimeMillis();
        
        //3. 根据考评指标创建相应的考评指标对象
        //将每一条考评指标都创建一个类：指标代码+Assessor
//        List<GovernanceAssessDetail> governanceAssessDetailList = new ArrayList<>();
        List<CompletableFuture<GovernanceAssessDetail>> governanceAssessDetailList = new ArrayList<>();
        for (TableMetaInfo tableMetaInfo : tableMetaInfoList) {
            for (GovernanceMetric governanceMetric : governanceMetricList) {
//                //反射方式
//                //约定：
//                //  包名为governance_type值
//                //  类名为metric_code值+Assessor
//                String basePackage = "com.atguigu.dga.governance.assessor";
//                String subPackage = governanceMetric.getGovernanceType().toLowerCase();
//                String fullPackage = basePackage + "." + subPackage;
//                
//                String metricCode = governanceMetric.getMetricCode();
//                String classNameByMetricCode = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.UPPER_CAMEL, metricCode);
//                String className = classNameByMetricCode + "." + "Assessor";
//                
//                try {
//                    Class<?> cls = Class.forName(className);
//                    Assessor assessor = (Assessor)cls.newInstance();
//                } catch (Exception e) {
//                    throw new RuntimeException(e);
//                }
            
                //springIOC容器方式
                //将考评指标类交给springIOC容器管理
                Assessor assessor = springIOCProvider.getByName(governanceMetric.getMetricCode(), Assessor.class);

                //获取任务定义和任务实例的key：schemaName.tableName
                String key = tableMetaInfo.getSchemaName() + "." + tableMetaInfo.getTableName();
                //考评参数
                AssessParam assessParam = AssessParam.builder()
                        .tableMetaInfo(tableMetaInfo)
                        .governanceMetric(governanceMetric)
                        .assessDate(assessDate)
                        .tableMetaInfoList(tableMetaInfoList)
                        .tDsTaskDefinition(tDsTaskDefinitionMap.get(key))
                        .tDsTaskInstance(tDsTaskInstanceMap.get(key))
                        .build();
                
                //处理白名单
                String skipAssessTables = governanceMetric.getSkipAssessTables();
                if (skipAssessTables != null && !skipAssessTables.trim().isEmpty()) {
                    String[] split = skipAssessTables.split(",");
                    //白名单表集合
                    List<String> list = Arrays.asList(split);
                    //跳过白名单中的表
                    if (list.contains(tableMetaInfo.getTableName())){
                        continue;
                    }
                }

                //考评
//                GovernanceAssessDetail governanceAssessDetail = assessor.doAssess(assessParam);
                
                //通过CompletableFuture.supplyAsync()方法 实现并发执行考评
                CompletableFuture<GovernanceAssessDetail> governanceAssessDetailFuture = CompletableFuture.supplyAsync(
                        () -> {
                            return assessor.doAssess(assessParam);
                        }, 
                        //线程池
                        threadPoolExecutor
                );

                //攒批
                governanceAssessDetailList.add(governanceAssessDetailFuture);
            }
        }
        //统一处理
        List<GovernanceAssessDetail> list = governanceAssessDetailList.stream().map(CompletableFuture::join).collect(Collectors.toList());
        saveBatch(list);

        long end = System.currentTimeMillis();
        System.out.println("执行时间：" + (end - start));
    }

    /**
     * 问题列表
     * @param governType
     * @param pageNo
     * @param pageSize
     * @return
     */
    @Override
    public String getProblemList(String governType, Integer pageNo, Integer pageSize) {
        int start = pageSize * (pageNo - 1);
        List<GovernanceAssessDetail> assessDetailList = list(
                new QueryWrapper<GovernanceAssessDetail>()
                        .inSql("assess_date", "SELECT MAX(assess_date) FROM governance_assess_detail")
                        .eq("governance_type", governType)
                        .lt("assess_score", 10)
                        .last("limit " + start + ", " + pageSize)
        );
        return JSONObject.toJSONString(assessDetailList);
    }

    /**
     * 各个治理类型问题个数
     * 返回值 {"SPEC":1, "STORAGE":4,"CALC":12,"QUALITY":34,"SECURITY":12}
     * @return
     */
    @Override
    public String getProblemNum() {
        //返回的结果格式为：list={{count(*): 153, governance_type: SPEC}, {count(*): 82, governance_type: SECURITY} ... }
        List<Map<String, Object>> list = listMaps(
                new QueryWrapper<GovernanceAssessDetail>()
                        .select("governance_type", "COUNT(*)")
                        .lt("assess_score", 10)
                        .inSql("assess_date", "SELECT MAX(assess_date) FROM governance_assess_detail")
                        .groupBy("governance_type")
        );

        Map<String, Long> resultMap = new HashMap<>();
        for (Map<String, Object> map : list) {
            resultMap.put(map.get("governance_type").toString(), (Long) map.get("COUNT(*)"));
        }
        
        return JSONObject.toJSONString(resultMap);
    }
}

